{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 3 - Data Lake on S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "import os\n",
    "import configparser"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Make sure that your AWS credentials are loaded as env vars"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "config = configparser.ConfigParser()\n",
    "\n",
    "#Normally this file should be in ~/.aws/credentials\n",
    "config.read_file(open('aws/credentials.cfg'))\n",
    "\n",
    "os.environ[\"AWS_ACCESS_KEY_ID\"]= config['AWS']['AWS_ACCESS_KEY_ID']\n",
    "os.environ[\"AWS_SECRET_ACCESS_KEY\"]= config['AWS']['AWS_SECRET_ACCESS_KEY']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create spark session with hadoop-aws package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder\\\n",
    "                     .config(\"spark.jars.packages\",\"org.apache.hadoop:hadoop-aws:2.7.0\")\\\n",
    "                     .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load data from S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"s3a://udacity-dend/pagila/payment/payment.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: string (nullable = true)\n",
      "\n",
      "+--------------------+\n",
      "|                 _c0|\n",
      "+--------------------+\n",
      "|payment_id;custom...|\n",
      "|16050;269;2;7;1.9...|\n",
      "|16051;269;1;98;0....|\n",
      "|16052;269;2;678;6...|\n",
      "|16053;269;2;703;0...|\n",
      "+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Infer schema, fix header and separator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(\"s3a://udacity-dend/pagila/payment/payment.csv\",sep=\";\", inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- payment_id: integer (nullable = true)\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- staff_id: integer (nullable = true)\n",
      " |-- rental_id: integer (nullable = true)\n",
      " |-- amount: double (nullable = true)\n",
      " |-- payment_date: string (nullable = true)\n",
      "\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|payment_id|customer_id|staff_id|rental_id|amount|        payment_date|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|     16050|        269|       2|        7|  1.99|2017-01-24 21:40:...|\n",
      "|     16051|        269|       1|       98|  0.99|2017-01-25 15:16:...|\n",
      "|     16052|        269|       2|      678|  6.99|2017-01-28 21:44:...|\n",
      "|     16053|        269|       2|      703|  0.99|2017-01-29 00:58:...|\n",
      "|     16054|        269|       1|      750|  4.99|2017-01-29 08:10:...|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Fix the data yourself "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- payment_id: integer (nullable = true)\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- staff_id: integer (nullable = true)\n",
      " |-- rental_id: integer (nullable = true)\n",
      " |-- amount: double (nullable = true)\n",
      " |-- payment_date: timestamp (nullable = true)\n",
      "\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|payment_id|customer_id|staff_id|rental_id|amount|        payment_date|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|     16050|        269|       2|        7|  1.99|2017-01-24 23:40:...|\n",
      "|     16051|        269|       1|       98|  0.99|2017-01-25 17:16:...|\n",
      "|     16052|        269|       2|      678|  6.99|2017-01-28 23:44:...|\n",
      "|     16053|        269|       2|      703|  0.99|2017-01-29 02:58:...|\n",
      "|     16054|        269|       1|      750|  4.99|2017-01-29 10:10:...|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import  pyspark.sql.functions as F\n",
    "dfPayment = df.withColumn(\"payment_date\", F.to_timestamp(\"payment_date\"))\n",
    "dfPayment.printSchema()\n",
    "dfPayment.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Extract the month"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+--------+---------+------+--------------------+-----+\n",
      "|payment_id|customer_id|staff_id|rental_id|amount|        payment_date|month|\n",
      "+----------+-----------+--------+---------+------+--------------------+-----+\n",
      "|     16050|        269|       2|        7|  1.99|2017-01-24 23:40:...|    1|\n",
      "|     16051|        269|       1|       98|  0.99|2017-01-25 17:16:...|    1|\n",
      "|     16052|        269|       2|      678|  6.99|2017-01-28 23:44:...|    1|\n",
      "|     16053|        269|       2|      703|  0.99|2017-01-29 02:58:...|    1|\n",
      "|     16054|        269|       1|      750|  4.99|2017-01-29 10:10:...|    1|\n",
      "+----------+-----------+--------+---------+------+--------------------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfPayment = dfPayment.withColumn(\"month\", F.month(\"payment_date\"))\n",
    "dfPayment.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Computer aggregate revenue per month"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------------+\n",
      "|month|           revenue|\n",
      "+-----+------------------+\n",
      "|    4|28094.520000003773|\n",
      "|    3|23886.560000002115|\n",
      "|    2| 9631.879999999608|\n",
      "|    1| 4824.429999999856|\n",
      "|    5| 979.1200000000023|\n",
      "+-----+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfPayment.createOrReplaceTempView(\"payment\")\n",
    "spark.sql(\"\"\"\n",
    "    SELECT month, sum(amount) as revenue\n",
    "    FROM payment\n",
    "    GROUP by month\n",
    "    order by revenue desc\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Fix the schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType as R, StructField as Fld, DoubleType as Dbl, StringType as Str, IntegerType as Int, DateType as Date\n",
    "paymentSchema = R([\n",
    "    Fld(\"payment_id\",Int()),\n",
    "    Fld(\"customer_id\",Int()),\n",
    "    Fld(\"staff_id\",Int()),\n",
    "    Fld(\"rental_id\",Int()),\n",
    "    Fld(\"amount\",Dbl()),\n",
    "    Fld(\"payment_date\",Date()),\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfPaymentWithSchema = spark.read.csv(\"s3a://udacity-dend/pagila/payment/payment.csv\",sep=\";\", schema=paymentSchema, header=True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- payment_id: integer (nullable = true)\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- staff_id: integer (nullable = true)\n",
      " |-- rental_id: integer (nullable = true)\n",
      " |-- amount: double (nullable = true)\n",
      " |-- payment_date: date (nullable = true)\n",
      "\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|payment_id|customer_id|staff_id|rental_id|amount|        payment_date|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "|     16050|        269|       2|        7|  1.99|2017-01-24 21:40:...|\n",
      "|     16051|        269|       1|       98|  0.99|2017-01-25 15:16:...|\n",
      "|     16052|        269|       2|      678|  6.99|2017-01-28 21:44:...|\n",
      "|     16053|        269|       2|      703|  0.99|2017-01-29 00:58:...|\n",
      "|     16054|        269|       1|      750|  4.99|2017-01-29 08:10:...|\n",
      "+----------+-----------+--------+---------+------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfPaymentWithSchema.printSchema()\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------------+\n",
      "|  m|           revenue|\n",
      "+---+------------------+\n",
      "|  4|28559.460000003943|\n",
      "|  3|23886.560000002115|\n",
      "|  2| 9631.879999999608|\n",
      "|  1| 4824.429999999856|\n",
      "|  5|  514.180000000001|\n",
      "+---+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfPaymentWithSchema.createOrReplaceTempView(\"payment\")\n",
    "spark.sql(\"\"\"\n",
    "    SELECT month(payment_date) as m, sum(amount) as revenue\n",
    "    FROM payment\n",
    "    GROUP by m\n",
    "    order by revenue desc\n",
    "\"\"\").show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
